#include "base.h"
#include "mpz_tool.h"
#include "basic_tool.h"
#include "config.h"
#include "crypto_tool.h"
#include "sql_parser.h"
#include "client_udp_wrapper.h"

void* ReceivingThread (void* arg)
{
  ClientUdpWrapper* obj = (ClientUdpWrapper* )arg;
  obj->ReceivingThread();
  return 0;
}

void* DecryptingThread (void* arg)
{
  ClientUdpWrapper* obj = (ClientUdpWrapper* )arg;
  obj->DecryptingThread();
  return 0;
}

void* OutputingThread (void* arg)
{
  ClientUdpWrapper* obj = (ClientUdpWrapper* )arg;
  obj->OutputingThread();
  return 0;
}


ClientUdpWrapper::ClientUdpWrapper()
{
  m_pServer = m_pIB = NULL;
  m_usCurrentRequestId = 0;
  pthread_mutex_init(&m_mutexNewUdpOrExit, NULL);
  pthread_mutex_init(&m_mutexOutput, NULL);
  pthread_mutex_init(&m_mutexRecord, NULL);
  pthread_cond_init (&m_cvNewUdpOrExit, NULL);
  pthread_cond_init (&m_cvOutput, NULL);
  pthread_cond_init (&m_cvRecord, NULL);
  m_N = Memreuse::New();
  m_g = Memreuse::New();
  m_e = Memreuse::New();
  m_p = Memreuse::New();
  m_q = Memreuse::New();
  m_y = Memreuse::New();
  m_p_minus_1_over_q = Memreuse::New();
  m_port = Config::m_client_port;
  m_strSigCounter.reserve(24);
  // read public keys
  if (Config::m_signature_protocol == "rsa")
    {
      string strRSAPubFile = Config::m_key_dir + "rsa.pub";
      BasicTool::Load_RSA_Pub_Key(m_e, m_N, m_g, strRSAPubFile);

      // counting the length of m_N
      size_t keylen = mpz_sizeinbase(*m_N, 2);
      m_rand.Initialize(keylen);
    }
  else if (Config::m_signature_protocol == "dsa")
    {
      string strDSAPubFile = Config::m_key_dir + "dsa.pub";
      BasicTool::Load_DSA_Pub_Key(m_p, m_q, m_g, m_y, strDSAPubFile);

      // counting the length of m_N
      size_t keylen = 160;
      m_rand.Initialize(keylen);
	  mpz_sub_ui(*m_p_minus_1_over_q, *m_p, 1);
	  mpz_divexact(*m_p_minus_1_over_q, *m_p_minus_1_over_q, *m_q);
    }
  else if (Config::m_signature_protocol == "ibe")
    {
      string strIBEClientFile = Config::m_key_dir + "ibe.client";
	  m_pairing = (pairing_t *) malloc(sizeof(pairing_t)*1);
	  m_emtP = (element_t * ) malloc(sizeof(element_t) * 1);
	  m_emtQ = (element_t * ) malloc(sizeof(element_t) * 1);
	  m_emts = (element_t * ) malloc(sizeof(element_t) * 1);
	  BasicTool::Load_IBE_Client_Key(m_pairing, m_emtP, m_emtQ, m_emts, strIBEClientFile);
      Memreuse::InitializeElements(1000, m_pairing);
    }

#ifdef _MEASURE
  ofCQC.open("cqc.log", ios_base::app);
  ofILS.open("ils.log", ios_base::app);
  ofRDD.open("rdd.log", ios_base::app);
  ofBand.open("band.log", ios_base::app);
#endif
}

ClientUdpWrapper::~ClientUdpWrapper()
{
  if (m_pServer)
	delete m_pServer;
  if (m_pIB)
	delete m_pIB;
  Memreuse::Delete(m_N);
  Memreuse::Delete(m_g);
  Memreuse::Delete(m_e);
  Memreuse::Delete(m_p);
  Memreuse::Delete(m_q);
  Memreuse::Delete(m_y);

#ifdef _MEASURE
  ofCQC.close();
  ofILS.close();
  ofRDD.close();
  ofBand.close();
#endif

}

void ClientUdpWrapper::ConnectServer(string serverip, unsigned short serverport)
{
  m_pServer = new UdpAgent();
  m_pServer->Connect(serverip, serverport);
  m_pServer->SetNonBlocking();
}

void ClientUdpWrapper::ConnectIB(string IBip, unsigned short IBport)
{
  bzero(&m_IBaddr, sizeof(m_IBaddr));
  m_IBaddr.sin_family = AF_INET;
  m_IBaddr.sin_port = htons(IBport);
  m_IBaddr.sin_addr.s_addr = inet_addr(IBip.c_str());
}

int ClientUdpWrapper::RecvSigsFromServer(size_t n, vector<string >& vecsig)
{
  UdpSigResponse * pRes = new UdpSigResponse;
  vecsig.clear();
  size_t i = 0;
  while (i < n)
	{
      while (!m_pServer->IsReadable(1000000)) // wait every 1s
		{
		  // resend query
		  UdpSigRequest * pReq = m_vecSigs[i].pReq;
		  m_pServer->Send((char *)pReq, sizeof(UdpSigRequest));
          cerr<<"resend "<<i<<"th sig request with subrequestid="<<pReq->SubRequestId()<<endl;
		}

      int n;
      n = m_pServer->Recv((char *)pRes, sizeof(UdpSigResponse));

	  // discard unordered response
	  if (pRes->RequestId() != m_usCurrentRequestId || pRes->SubRequestId() != i)
        {
          cerr<<"receive subrequestid="<<pRes->SubRequestId()<<endl;
          continue;
        }
	  string strSig;
      if (Config::m_signature_protocol == "rsa")
        {
		  mpz_t * mpz_blindsig = Memreuse::New();
		  mpz_t * mpz_invr = m_vecSigs[i].mpzRandInv;
		  MpzTool::Mpz_from_char(mpz_blindsig, pRes->GetSig(), pRes->SigLength());
		  // recover blind signature
		  mpz_mul(*mpz_blindsig, *mpz_blindsig, *mpz_invr);

          mpz_mod(*mpz_blindsig, *mpz_blindsig, *m_N);
		  strSig = MpzTool::String_from_mpz(NULL, mpz_blindsig);
		  strSig = CryptoTool::Hash(strSig);
		  vecsig.push_back(strSig);

		  Memreuse::Delete(mpz_invr);
		  Memreuse::Delete(mpz_blindsig);
        }
      else if (Config::m_signature_protocol == "dsa")
        {
		  mpz_t * mpz_blindsig = Memreuse::New();
		  mpz_t * mpz_invr = m_vecSigs[i].mpzRandInv;
		  MpzTool::Mpz_from_char(mpz_blindsig, pRes->GetSig(), pRes->SigLength());
		  // recover blind signature
		  mpz_mul(*mpz_blindsig, *mpz_blindsig, *mpz_invr);
          mpz_mod(*mpz_blindsig, *mpz_blindsig, *m_p);
		  strSig = MpzTool::String_from_mpz(NULL, mpz_blindsig);
		  strSig = CryptoTool::Hash(strSig);
		  vecsig.push_back(strSig);

		  Memreuse::Delete(mpz_invr);
		  Memreuse::Delete(mpz_blindsig);
        }
      else if (Config::m_signature_protocol == "ibe")
        {
		  element_t * emtR = Memreuse::NewG1Element();
		  element_t * emtT = Memreuse::NewGTElement();
		  element_from_bytes(*emtR, (unsigned char *)pRes->GetSig());
		  element_t * emtSig = m_vecSigs[i].emtSig;
		  pairing_apply(*emtT, *emtR, *emtSig, *m_pairing);
		  string strSig = MpzTool::Element_to_string(emtT);
		  strSig = CryptoTool::Hash(strSig);
		  vecsig.push_back(strSig);

		  Memreuse::DeleteG1Element(emtR);
		  Memreuse::DeleteGTElement(emtT);
        }
	  delete m_vecSigs[i].pReq;	// release space
      i++;
	}
  m_vecSigs.clear();
  delete pRes;
  return 0;
}

int ClientUdpWrapper::SendSigRequestsToServer(const vector<struct tagCommand>& vecSelectArgs)
{

  m_vecSigs.clear();

  for (size_t i = 0; i < vecSelectArgs.size(); i++)
	{
	  UdpSigRequest * pReq = new UdpSigRequest;
	  mpz_t * mpzHashAtVal = Memreuse::New();
      mpz_t * mpz_r = Memreuse::New();
      if (Config::m_signature_protocol == "rsa")
        {
		  string strHashAtVal = CryptoTool::Hash(vecSelectArgs[i].strAtVal);
		  MpzTool::Mpz_from_string(mpzHashAtVal, strHashAtVal);
          // compute r coprime to N
          mpz_t * mpz_invr = Memreuse::New();
          m_rand.Get_Z_Range(mpz_r, m_N);

          while (!mpz_invert(*mpz_invr, *mpz_r, *m_N))
            {
              m_rand.Get_Z_Range(mpz_r, m_N);
            }
          mpz_powm(*mpz_r, *mpz_r, *m_e, *m_N);
          mpz_mul(*mpzHashAtVal, *mpzHashAtVal, *mpz_r);
          mpz_mod(*mpzHashAtVal, *mpzHashAtVal, *m_N);

          pReq->RequestId() = m_usCurrentRequestId;
          pReq->SubRequestId() = i;
          pReq->SetSig(MpzTool::String_from_mpz(NULL, mpzHashAtVal));
          m_pServer->Send((char *)pReq, sizeof(UdpSigRequest));

          struct tagSig ssig;
          ssig.pReq = pReq;
          ssig.mpzRandInv = mpz_invr;
          m_vecSigs.push_back(ssig);

        }
      else if (Config::m_signature_protocol == "dsa")
        {
          CryptoTool::FDH(mpzHashAtVal, vecSelectArgs[i].strAtVal, m_p);
		  mpz_powm(*mpzHashAtVal, *mpzHashAtVal, *m_p_minus_1_over_q, *m_p);

          mpz_t * t1 = Memreuse::New();
          m_rand.Get_Z_Range(mpz_r, m_q);
          mpz_powm(*t1, *m_g, *mpz_r, *m_p); // t1=g^r
          mpz_mul(*mpzHashAtVal, *mpzHashAtVal, *t1);
          mpz_mod(*mpzHashAtVal, *mpzHashAtVal, *m_p); // mpzHashAtVal=hash^((p-1)/q)*g^r
          mpz_invert(*t1, *m_y, *m_p);                 // t1=y^-1
          mpz_powm(*t1, *t1, *mpz_r, *m_p);            // t1=y^-r

          pReq->RequestId() = m_usCurrentRequestId;
          pReq->SubRequestId() = i;
          pReq->SetSig(MpzTool::String_from_mpz(NULL, mpzHashAtVal));
          m_pServer->Send((char *)pReq, sizeof(UdpSigRequest));

          struct tagSig ssig;
          ssig.pReq = pReq;
          ssig.mpzRandInv = t1;
          m_vecSigs.push_back(ssig);
        }
      else if (Config::m_signature_protocol == "ibe")
        {
		  element_t * emt_h = Memreuse::NewG2Element();
		  element_t * emt_sig = Memreuse::NewG2Element();
		  string h = CryptoTool::Hash(vecSelectArgs[i].strAtVal);
		  pReq->RequestId() = m_usCurrentRequestId;
          pReq->SubRequestId() = i;
          m_pServer->Send((char *)pReq, sizeof(UdpSigRequest));
		  MpzTool::Element_from_string(emt_h, h);
		  element_pow_zn(*emt_sig, *emt_h, *m_emts);

          struct tagSig ssig;
          ssig.pReq = pReq;
		  ssig.emtSig = emt_sig;
          m_vecSigs.push_back(ssig);
		  Memreuse::DeleteG2Element(emt_h);
        }

	  Memreuse::Delete(mpzHashAtVal);
      Memreuse::Delete(mpz_r);
	}
  return 0;
}

int ClientUdpWrapper::SendRequestToIB(string strTag)
{
  UdpRequest * pReq = new UdpRequest;
  pReq->RequestId() = ++m_usCurrentRequestId;
  pReq->SubRequestId() = 0;
  pReq->BlockId() = -1;
  pReq->Limit() = m_limit;
  pReq->SetTag(strTag);
  pReq->TagId() = m_iTagId;
  //pReq->SetIndKey(strIndKey);
  m_pIB->SendTo((char *)pReq, sizeof(UdpRequest), m_IBaddr);
//   mpz_t * mpzTag = Memreuse::New();
//   MpzTool::Mpz_from_string(mpzTag, strTag);
//   gmp_printf("request hashki = %Zx\n", *mpzTag);
  delete pReq;
  return 0;
}

void ClientUdpWrapper::ReceiveOneBlock(UdpResponse * pUdp, struct sockaddr_in& from)
{
  while (!m_bThreadExit)
	{
	  if (m_pIB->IsReadable(1000))
		{
		  m_pIB->RecvFrom((char*)pUdp, sizeof(UdpResponse), from);
		  return;
		}
	}
}

void ClientUdpWrapper::ReceiveOneBlockFromIB(UdpResponse * pUdp)
{
  while (!m_bThreadExit)
	{
	  if (m_pIB->IsReadable(1000))
		{
		  m_pIB->Recv((char*)pUdp, sizeof(UdpResponse));
		  return;
		}
	}
}


void ClientUdpWrapper::CleanUp()
{
  m_sliding_window.CleanUp();
  while (!m_queueToOutput.empty())
	{
	  Record * pRec = m_queueToOutput.front();
      Memreuse::Delete(pRec);
	  m_queueToOutput.pop_front();
	}

  if (m_pIB)
	delete m_pIB;

  m_pIB = new UdpAgent();
  while (m_pIB->Bind(Config::m_client_ip, m_port++) < 0)
    {
      cerr<<"port "<<m_port-1<<" is in use"<<endl;
    }
  // bind to a new port
#ifdef _DEBUG
  cerr<<"bind to "<<m_port;
#endif
  m_pIB->SetNonBlocking();
  if (m_port >= 8888)
	{
	  m_port = Config::m_client_port;
	}
  m_iTagId = 0;
}

void ClientUdpWrapper::ReceivingThread()
{
  m_bFirstResp = true;
  struct sockaddr_in from;
  while (1)
    {
      UdpResponse * pUdp = new UdpResponse;
	  if (m_bFirstResp)
		{
		  ReceiveOneBlock (pUdp, from);
		}
	  else
		{
		  ReceiveOneBlockFromIB(pUdp);
		}
	  // signal if exit
	  if (m_bThreadExit)
		{
		  pthread_mutex_lock(&m_mutexNewUdpOrExit);

		  pthread_cond_signal(&m_cvNewUdpOrExit);
		  
		  pthread_mutex_unlock(&m_mutexNewUdpOrExit);
		  
		  break;
		}
	  else	  // signal if new packet
		{
		  if (pUdp->ResponseId() != m_usCurrentRequestId)
			{
			  cerr<<"out of bound response "<<pUdp->ResponseId()<<":"<<pUdp->ID()<<endl;
			  cerr<<"current request id is "<<m_usCurrentRequestId<<endl;
			  delete pUdp;
			  continue;				// if it is the response to the last request
			}
#ifdef _DEBUG
          cerr<<BasicTool::TimeStamp()<<" receive id = "<<pUdp->ID()<<" receive rid = "<<pUdp->SubResponseId()<<" bid = "<<pUdp->BlockId()<<endl;
#endif

		  pthread_mutex_lock(&m_mutexNewUdpOrExit);

          if (!m_sliding_window.Enqueue(pUdp))
            {
              delete pUdp;
            }
		  else
			{
			  if (m_bFirstResp)
				{
				  m_bFirstResp = false;
				  m_pIB->Connect(from);
#ifdef _MEASURE
				  ofILS<<BasicTool::TimeStamp()<<" "<<BasicTool::F_Time(BasicTool::STOP)<<endl;
				  BasicTool::F_Time(BasicTool::START);
#endif
				}
			}

		  pthread_cond_signal(&m_cvNewUdpOrExit);
		  
		  pthread_mutex_unlock(&m_mutexNewUdpOrExit);
		}
    }
  pthread_exit(NULL);
}

// send request for first unreceived block
void ClientUdpWrapper::Ack()
{
  if (m_bFirstResp)				// if no response ever received
	{
	  // resend first request
	  SendRequestToIB(m_strTag);
      cerr<<"resending request"<<endl;
#ifdef _DEBUG
      cerr<<"resending request"<<endl;
#endif
	}
  else
    {
      // request the first block inside sliding window
      UdpAck * pAck = new UdpAck;
      pAck->ID() = m_sliding_window.HeadId(); // request for this id which is the next id to be processed
      pAck->Capacity() = m_sliding_window.m_bitmap;
	  pAck->SetTag(m_strTag);
      pAck->TagId() = m_iTagId;
      //	  pAck->SetIndKey(m_strIndKey);
      m_pIB->Send((char *)pAck, sizeof(UdpAck));
#ifdef _DEBUG
      cerr<<BasicTool::TimeStamp()<<" ack id = "<<pAck->ID()<<" capacity="<<pAck->Capacity()<<endl;
#endif
	  delete pAck;
    }
}

void ClientUdpWrapper::DecryptingThread()
{
  struct timespec to;
  int err;
  static RC4_KEY rc_key;
  vector<UdpResponse *> vecpUdp;
  while (1)
    {
      UdpResponse * pUdp = NULL;
      vecpUdp.clear();          // delete memory is not necessary
      pthread_mutex_lock(&m_mutexNewUdpOrExit);
      // put all packets into vecpUdp
      while (!m_bThreadExit)
        {
          pUdp = m_sliding_window.Front();
          if (pUdp == NULL && vecpUdp.size() > 0)
            break;

          while (pUdp == NULL && !m_bThreadExit)
            {
              clock_gettime(CLOCK_REALTIME, &to);
              to.tv_sec += 1;
              err = pthread_cond_timedwait(&m_cvNewUdpOrExit, &m_mutexNewUdpOrExit, &to);
              if (err == ETIMEDOUT)
                {
                  // retransmit request
                  cerr<<BasicTool::TimeStamp()<<" receive timeout"<<endl;
                }
              pUdp = m_sliding_window.Front();
              if (pUdp == NULL)
                {
                  // if new packet is not at the head or timeout
                  Ack();
                }
            }
          if (pUdp != NULL && !m_bThreadExit)
            {
              m_sliding_window.Pop();
              vecpUdp.push_back(pUdp);
            }
        }
      pthread_mutex_unlock(&m_mutexNewUdpOrExit);

	  if (m_bThreadExit)
		{
		  return;
		}
      Ack();                    // ack once
      vector<UdpResponse *>::iterator it;
      for (it = vecpUdp.begin(); it != vecpUdp.end(); ++it)
        {
          pUdp = *it;
          if (pUdp->BlockId() == -1) // key or end of response
            {
              clock_gettime(CLOCK_REALTIME, &to);
              to.tv_sec += 1;
              pthread_mutex_lock(&m_mutexRecord);
              while (Memreuse::RecordSize() == 0 && !m_bThreadExit)
                {
                  err = pthread_cond_timedwait(&m_cvRecord, &m_mutexRecord, &to);
                  if (err == ETIMEDOUT)
                    {
                      to.tv_sec += 1;
                      cerr<<BasicTool::TimeStamp()<<" newrecord operation timeout"<<endl;
                    }
                  Ack();				// prevent server timeout
                }
              if (!m_bThreadExit)
                m_pRecord = Memreuse::NewRecord();
              pthread_mutex_unlock(&m_mutexRecord);

              if (m_bThreadExit)	// possible if connection end
                {
                  return;
                }

              if (pUdp->TotalBlocks() >(int)( m_pRecord->m_blocks.size()))
                {
                  m_pRecord->m_blocks.resize(pUdp->TotalBlocks(), NULL);
                }
              m_pRecord->m_iTotalBlocks = pUdp->TotalBlocks();
          
              if (pUdp->TotalBlocks() > 0) // key 
                {

                  // m_strIndKey = CryptoTool::HIndKey(ss.str());
                  string strEncK_Ki;
                  strEncK_Ki.assign(pUdp->GetBlock(), pUdp->BlockLength());
                  string strKey = CryptoTool::XOR(strEncK_Ki, m_strKi);

                  ::RC4_set_key(&rc_key, strKey.size(), (const unsigned char *)(strKey.c_str()));
                  // memcpy(m_cSigCounter, m_strSig.c_str(), HASH_LEN); //does not need
                  memcpy(m_cSigCounter+HASH_LEN, (const char *)&m_iCounter, sizeof(int));
                  m_strSigCounter.assign(m_cSigCounter, 24);
                  m_iCounter++;
                  m_strTag = CryptoTool::HTag(m_strSigCounter); // update tag
                  m_iTagId ++;
                  m_strKi = CryptoTool::HKey(m_strSigCounter); // update key
                }
              else                  // end of response
                {
                  pthread_mutex_lock(&m_mutexOutput);
                  m_queueToOutput.push_back(m_pRecord);
                  pthread_cond_signal(&m_cvOutput);
                  pthread_mutex_unlock(&m_mutexOutput);
#ifdef _DEBUG
                  cerr<<"end of response"<<endl;
#endif
                }
            }
          else
            {
              int blockid = pUdp->BlockId();
              // decrypt
              if (m_pRecord->m_blocks[blockid] == NULL)
                {
                  m_pRecord->m_blocks[blockid] = Memreuse::NewBlock();
                }
              if (blockid < m_pRecord->m_iTotalBlocks)
                {
                  char * buf = m_pRecord->m_blocks[blockid]->GetBuf();
                  size_t& len = m_pRecord->m_blocks[blockid]->Length();
                  len = pUdp->BlockLength();
                  ::RC4(&rc_key, len,(const unsigned char *)pUdp->GetBlock(), 
                        (unsigned char *)buf);
                }
              if (blockid == m_pRecord->m_iTotalBlocks - 1)
                {
                  pthread_mutex_lock(&m_mutexOutput);
                  m_queueToOutput.push_back(m_pRecord);
                  pthread_cond_signal(&m_cvOutput);
                  pthread_mutex_unlock(&m_mutexOutput);
                }
            }
          delete pUdp;
          //Ack();
        }
	}
  pthread_exit(NULL);
}

void ClientUdpWrapper::OutputingThread()
{
  struct timespec to;
  int err;
  Record * pRecord;
  while (1)
    {
      pthread_mutex_lock(&m_mutexOutput);
      if (m_queueToOutput.empty())
        {
          //cerr<<"m_queueToOutput.empty()"<<endl;
		  clock_gettime(CLOCK_REALTIME, &to);
          to.tv_sec += 10;
          err = pthread_cond_timedwait(&m_cvOutput, &m_mutexOutput, &to);
		  if (err == ETIMEDOUT)
			{
			  pRecord = NULL;
			  m_bThreadExit = true; // without input for 10s
              cerr<<"output timeout"<<endl;
			}
        }
	  if (!m_queueToOutput.empty())
		{
		  pRecord = m_queueToOutput.front();
		  m_queueToOutput.pop_front();
		}
      pthread_mutex_unlock(&m_mutexOutput);

	  if (m_bThreadExit)
		{
		  cerr<<"connection end"<<endl;
		  break;
		}
	  // if everything is done
	  if (pRecord->IsEndofResponse())
		{
          if (!m_bBatch)
            cout<<"="<<flush;
#ifdef _MEASURE
		  cerr<<"finished "<<m_iResponses<<" records"<<endl;
#endif
		  m_bThreadExit = true;
		  pthread_mutex_lock(&m_mutexRecord);
          Memreuse::Delete(pRecord);
		  pthread_cond_signal(&m_cvRecord);
		  pthread_mutex_unlock(&m_mutexRecord);
		  break;
		}

      bool bDup = false;
      int outputedlen = 0;
	  for (int i = 0; i < pRecord->m_iTotalBlocks; i++)
		{
		  char * buf = pRecord->m_blocks[i]->GetBuf();
		  size_t len = pRecord->m_blocks[i]->Length();
		  if (m_bBatch && i == 0)
			{
			  // don't output duplicates in batch mode
			  string h = CryptoTool::Hash(buf, len/2);
			  if (m_hashMap.find(h) == m_hashMap.end())
				{
				  m_hashMap[h] = true;
				}
			  else
				{
                  bDup = true;
				  break;		// the record was outputted before
				}
			}
		  if (len > 0)
			{
			  if (!Config::m_output_binaryfile && outputedlen >= 1024)
				{
				  // do nothing
				}
			  else
				{
				  cout.write(buf, len); 
				}
			  outputedlen += len;
#ifdef _MEASURE
              m_iBytes += len;
#endif
			  len = 0;
			}
          //Memreuse::Delete(pRecord->m_blocks[i]); keep blocks allocated in record
		}
	  pthread_mutex_lock(&m_mutexRecord);
	  Memreuse::Delete(pRecord);
	  pthread_cond_signal(&m_cvRecord);
	  pthread_mutex_unlock(&m_mutexRecord);
      if (!bDup)
        cout<<"-"<<flush;
#ifdef _MEASURE
      m_iResponses++;
#endif
    }
  pthread_exit(NULL);
}

void ClientUdpWrapper::Execute(vector<struct tagCommand> vecSelectArgs)
{
#ifdef _MEASURE
  BasicTool::F_Time(BasicTool::START);
#endif
  SendSigRequestsToServer(vecSelectArgs);
  vector<string > vecSig;
  RecvSigsFromServer(vecSelectArgs.size(), vecSig);
  m_bBatch = false;
  if (vecSig.size() > 1)
    {
      m_bBatch = true;
    }

  pthread_t tRecv, tDec, tOutput;
  for (size_t i = 0; i < vecSig.size(); i++)
	{
	  m_firstUnreceived.m_iRespId = 0;
	  m_firstUnreceived.m_iBlockId = -1;

	  m_limit = vecSelectArgs[i].limit;
	  m_offset = vecSelectArgs[i].offset;
      if (m_offset > 0)
        m_iCounter = m_offset;
      else 
        m_iCounter = 0;
      m_strSig = vecSig[i];
      memcpy(m_cSigCounter, m_strSig.c_str(), HASH_LEN);
      memcpy(m_cSigCounter+HASH_LEN, (const char *)&m_iCounter, sizeof(int));
      m_strSigCounter.assign(m_cSigCounter, 24);
	  m_iCounter++;
	  m_strKi = CryptoTool::HKey(m_strSigCounter);
	  //	  m_strIndKey = CryptoTool::HIndKey(ss.str());
	  m_strTag = CryptoTool::HTag(m_strSigCounter);

	  SendRequestToIB(m_strTag);

	  // create threads for reading and sending
	  m_bThreadExit = false;

	  pthread_create(&tRecv, NULL, ::ReceivingThread, (void*)this);
	  pthread_create(&tDec, NULL, ::DecryptingThread, (void*)this);
	  pthread_create(&tOutput, NULL, ::OutputingThread, (void*)this);

	  pthread_join(tRecv, NULL );
	  pthread_join(tDec, NULL );
	  pthread_join(tOutput, NULL );

	  CleanUp();
	}
  if (m_bBatch)
    {
      cout<<"="<<flush;
    }
}

int ClientUdpWrapper::Start()
{
  vector<struct tagCommand > vecSelectArgs;
  string line;
  string lowercaseline;
  string command;
  map<string, string> args;
  bool bBegin = true;
  bool bEnd = false;
  CleanUp();
  while(!bEnd)
	{
	  if (!bBegin)
		{
		  cout<<endl;
		}
	  else
		{
		  bBegin = false;
		}
	  cout<<">> "<<flush;

	  getline(cin, line);

#ifdef _MEASURE
      BasicTool::F_Time(BasicTool::START);
      ofCQC<<BasicTool::TimeStamp()<<" "<<BasicTool::F_Time(BasicTool::STOP)<<endl;
#endif
	  
	  SQLParser::Parse(line, command, args);
	  if (command == "end" || command == "done")
		{
		  bEnd = true;
		}
	  else if (command == "execute")
		{
#ifdef _MEASURE
          m_iResponses = 0;
          m_iBytes = 0;
#endif
          m_hashMap.clear();    // don't put this in the CleanUp()

          Execute(vecSelectArgs);
#ifdef _MEASURE
          ofRDD<<BasicTool::TimeStamp()<<" "<<BasicTool::F_Time(BasicTool::STOP)<<" "<<m_iResponses<<endl;
          ofBand<<BasicTool::TimeStamp()<<" "<<m_iBytes<<endl;
#endif          
          vecSelectArgs.clear();
		}
	  else if (command == "select")
		{
		  struct tagCommand opt;
		  opt.strAtVal = args["strAtVal"];
		  opt.offset = atoi(args["strOffset"].c_str());
		  opt.limit = atoi(args["strLimit"].c_str());
		  vecSelectArgs.push_back(opt);
		}
	}
  return 0;
}

